#pragma once

#include "../common/net.hpp"
#include "../common/message.hpp"
#include <unordered_set>

namespace JsonMuduoRPC
{
    namespace server
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;

            void onTopicRequest(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                bool ret = true;
                TopicOpType opType = msg->getOpType();

                switch (opType)
                {
                // 主题创建
                case TopicOpType::TOPIC_CREATE:
                    createTopic(conn, msg);
                    break;
                // 主题删除
                case TopicOpType::TOPIC_REMOVE:
                    removeTopic(conn, msg);
                    break;
                // 主题订阅
                case TopicOpType::TOPIC_SUBSCRIBE:
                    ret = subscribeTopic(conn, msg);
                    break;
                // 取消订阅
                case TopicOpType::TOPIC_CANCEL:
                    cancelTopic(conn, msg);
                    break;
                // 发布消息
                case TopicOpType::TOPIC_PUBLISH:
                    ret = publishTopic(conn, msg);
                    break;
                default:
                    return errorResponse(conn, msg, RCode::RCODE_INVALID_OPTYPE);
                }

                if (!ret)
                    return errorResponse(conn, msg, RCode::RCODE_NOT_FOUND_TOPIC);

                registryResponse(conn, msg);
            }

            // 订阅者关闭连接
            void onShutdown(const BaseConnection::ptr &conn)
            {
                std::vector<Topic::ptr> topics;
                Subscriber::ptr subscriber;
                {
                    // 首先找到订阅者
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it_subscribe = _subscribers_map.find(conn);
                    if (it_subscribe == _subscribers_map.end())
                        return; // 不是一个订阅者

                    subscriber = it_subscribe->second;

                    // 再将该订阅者订阅的主题对应的主题对象找到
                    for (auto &topic : subscriber->topics)
                    {
                        auto it_topic = _topics_map.find(topic);
                        if (it_topic != _topics_map.end())
                            topics.push_back(it_topic->second);
                    }

                    // 将订阅者连接的映射关系删除
                    _subscribers_map.erase(it_subscribe);
                }

                //将所有订阅过的主题对象中的该订阅者删除
                for (auto& topic : topics)
                    topic->removeSubscriver(subscriber);
            }

        private:
            // 错误响应
            void errorResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg, const RCode rcode)
            {
                TopicResponse::ptr rsp = MessageFactory::create<TopicResponse>();
                rsp->setId(msg->getId());
                rsp->setMType(MType::RSP_TOPIC);
                rsp->setRCode(rcode);

                conn->send(rsp);
            }

            // 主题响应
            void registryResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                TopicResponse::ptr rsp = MessageFactory::create<TopicResponse>();
                rsp->setId(msg->getId());
                rsp->setMType(MType::RSP_TOPIC);
                rsp->setRCode(RCode::RCODE_OK);

                conn->send(rsp);
            }

        private:
            // 主题创建
            void createTopic(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                const std::string &topic_name = msg->getTopic();

                std::unique_lock<std::mutex> lock(_mutex);
                Topic::ptr topic = std::make_shared<Topic>(topic_name);
                _topics_map.insert(std::make_pair(topic_name, topic));
            }

            // 主题删除
            void removeTopic(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                const std::string &topic_name = msg->getTopic();

                // 首先找到订阅了该主题的客户端，将该主题从客户端的订阅中删除
                std::unordered_set<Subscriber::ptr> subscribers;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it_topic = _topics_map.find(topic_name);
                    if (it_topic == _topics_map.end())
                    {
                        LOG(ERROR, "不存在主题: %s", topic_name.c_str());
                        return;
                    }

                    subscribers = it_topic->second->subscribers;
                    _topics_map.erase(it_topic); // 找到客户端后就将主题的删除
                }

                // 将该主题从客户端的订阅中删除
                for (auto &subscriber : subscribers)
                    subscriber->removeTopic(topic_name);
            }

            // 主题订阅
            bool subscribeTopic(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                Topic::ptr topic;
                Subscriber::ptr subscribe;

                {
                    // 找到对应的主题对象
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it_topic = _topics_map.find(msg->getTopic());
                    if (it_topic == _topics_map.end())
                    {
                        LOG(ERROR, "不存在主题: %s", msg->getTopic().c_str());
                        return false;
                    }
                    topic = it_topic->second;
                    // 找到连接对应的订阅者
                    // 不存在便创建，同时在_subscribers_map中构建映射关系
                    auto it_subscribe = _subscribers_map.find(conn);
                    if (it_subscribe == _subscribers_map.end())
                    {
                        subscribe = std::make_shared<Subscriber>(conn);
                        _subscribers_map.insert(std::make_pair(conn, subscribe));
                    }
                    else
                        subscribe = it_subscribe->second;
                }

                // 添加对应关系
                topic->appendSubscriber(subscribe);
                subscribe->appendTopic(msg->getTopic());

                return true;
            }

            // 取消订阅
            void cancelTopic(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                Topic::ptr topic;
                Subscriber::ptr subscribe;
                {
                    // 找到连接对应的订阅者
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it_subscribe = _subscribers_map.find(conn);
                    if (it_subscribe != _subscribers_map.end())
                        subscribe = it_subscribe->second;

                    // 找到对应的主题对象
                    auto it_topic = _topics_map.find(msg->getTopic());
                    if (it_topic != _topics_map.end())
                        topic = it_topic->second;
                }

                // 从主题中删除订阅者 从订阅者中删除主题
                if (topic && subscribe)
                    topic->removeSubscriver(subscribe);
                if (subscribe)
                    subscribe->removeTopic(msg->getTopic());
            }

            // 发布消息
            bool publishTopic(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                Topic::ptr topic;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _topics_map.find(msg->getTopic());
                    if (it == _topics_map.end())
                    {
                        LOG(ERROR, "推送的主题 %s 不存在", msg->getTopic().c_str());
                        return false;
                    }
                    topic = it->second;
                }

                topic->pushMessage(msg);
                return true;
            }

        private:
            struct Subscriber
            {
                using ptr = std::shared_ptr<Subscriber>;

                Subscriber(const BaseConnection::ptr &c) : conn(c) {}

                std::mutex mutex;
                BaseConnection::ptr conn;
                std::unordered_set<std::string> topics; // 这个客户端订阅了哪些主题

                // 订阅主题
                void appendTopic(const std::string &topic)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    topics.insert(topic);
                }

                // 取消订阅
                void removeTopic(const std::string &topic)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    topics.erase(topic);
                }
            };
            struct Topic
            {
                using ptr = std::shared_ptr<Topic>;

                Topic(const std::string &name) : topic_name(name) {}

                std::mutex mutex;
                std::string topic_name;
                std::unordered_set<Subscriber::ptr> subscribers; // 这个主题被哪些客户端订阅

                // 当一个客户端订阅了该主题后，就要将该客户端管理起来
                void appendSubscriber(const Subscriber::ptr subscriver)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    subscribers.insert(subscriver);
                }

                // 当一个客户端取消订阅了该主题，就要将管理的客户端删除
                void removeSubscriver(const Subscriber::ptr subscriver)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    subscribers.erase(subscriver);
                }

                // 将消息推送给每一个订阅了该主题的订阅者
                void pushMessage(const BaseMessage::ptr &msg)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    for (auto &subsriber : subscribers)
                        subsriber->conn->send(msg);
                }
            };

        private:
            std::mutex _mutex;
            std::unordered_map<std::string, Topic::ptr> _topics_map;                   // 主题——订阅者
            std::unordered_map<BaseConnection::ptr, Subscriber::ptr> _subscribers_map; // 连接——订阅者
        };
    }
}